package com.zyx.pulsardemo.basic.consumers;

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SubscriptionType;

/**
 * @author Yaxi.Zhang
 * @since 2021/9/30 16:52
 * desc: Pulsar生产者和消费者对, 无限生产和消费消息
 * reference: https://github.com/david-streamlio/pulsar-in-action
 */
public class BackAndForth extends PulsarConsumerDemoBase {

    public static void main(String[] args) {
        BackAndForth sl = new BackAndForth();
        // sl.startProducer();
        sl.startConsumer();
    }

    @Override
    protected Consumer<byte[]> getConsumer() {
        try {
            return getClient()
                    .newConsumer()
                    .topic(topic)
                    .subscriptionName(subscriptionName)
                    .subscriptionType(SubscriptionType.Exclusive)
                    .subscribe();
        } catch (PulsarClientException e) {
            return null;
        }
    }

}
